package com.rtsapp.server.network.protocol.rpc.client.impl;

import com.rtsapp.server.module.Module;
import com.rtsapp.server.network.protocol.rpc.client.RPCClient;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.*;
import io.netty.channel.nio.*;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import com.rtsapp.server.logger.Logger;

import com.rtsapp.server.module.AbstractModule;
import com.rtsapp.server.module.Application;

import java.util.*;


/**
 * RPC客户端模块, 实现了RPCClient接口
 * 
 */
public class RPCClientImpl extends AbstractModule implements RPCClient {
	
	private static final Logger logger =com.rtsapp.server.logger.LoggerFactory.getLogger( RPCClientImpl.class );

	//是否linux平台
	private boolean linux;
	//工作线程数量
	private int workerThread;

	private List<ClientCfg> clientCfgList;

	protected Application application;
	//事件循环
	protected EventLoopGroup clientEventLoop;

	//服务器名称-对应的连接处理列表
	private Map<String, RPCClientProxy > serverProxyMap = new java.util.concurrent.ConcurrentHashMap< String, RPCClientProxy >( );



	protected boolean isLinux(){
		return this.linux;
	}
	
	public void setLinux(boolean linux) {
		this.linux = linux;
	}

	public void setWorkerThread(int workerThread) {
		this.workerThread = workerThread;
	}

	public void setClientCfgList(List<ClientCfg> clientCfgList) {
		this.clientCfgList = clientCfgList;
	}


	@Override
	public boolean initialize(Module parentModule) {
	
		//创建事件循环组
		if( parentModule != null ) {
			this.application = (Application) parentModule;
		}


		if ( isLinux() ) {
			this.clientEventLoop = new EpollEventLoopGroup(this.workerThread,  new DefaultThreadFactory( this.getName() + ":thread" ));
		} else {
			this.clientEventLoop = new NioEventLoopGroup(this.workerThread , new DefaultThreadFactory( this.getName() + ":thread" ) );
		}


		if( clientCfgList != null ) {
			for (ClientCfg cfg : clientCfgList) {
				createClientProxy( cfg );
			}
		}

		//	
		return true;
	}

	/**
	 * 创建一个ClientProxy，但是不调用start
	 * @param cfg
	 * @return 如果同名的客户端代理已经存在, 则返回false
	 */
	private boolean createClientProxy( ClientCfg cfg ){

		if (!serverProxyMap.containsKey(cfg.getServerName())) {
			RPCClientProxy proxy = new RPCClientProxy(cfg, clientEventLoop, isLinux());
			serverProxyMap.put(cfg.getServerName(), proxy);
			return true;
		} else {
			logger.error("ClientCfg duplicate, server:" + cfg.getServerName() + " is already exist");
			return false;
		}

	}

	/**
	 * 服务启动后加入一个客户端连接
	 * 原有同名客户端连接，这个不生效
	 * @param cfg
	 */
	protected void addClientProxyAfterStart( ClientCfg cfg  ){
		if( createClientProxy( cfg ) ){
			RPCClientProxy proxy = getRPCClientProxy( cfg.getServerName() );
			if( proxy != null ){
				proxy.start();
			}
		}
	}
	
	@Override
	public boolean start() {

		for( Map.Entry<String, RPCClientProxy > entry : this.serverProxyMap.entrySet( ) ){
			entry.getValue().start( );
		}

		return true;
	}


	@Override
	public void stop() {

		Future future = this.clientEventLoop.shutdownGracefully( );
		try {
			future.await();
		} catch (InterruptedException e) {
			logger.error( " RPCClientImpl stop 被 InterruptedException", e );
		}

	}

	@Override
	public void destroy() {
		
	}


	@Override
	public Object call( String server,  String service, String funName, Object...args ){

		RPCClientProxy proxy = this.getRPCClientProxy( server );

		if( proxy != null ){
			return proxy.call( service, funName, args );
		}else{
			logger.error( "server not find , name=" + server );
			return null;
		}

	}

	public RPCClientProxy getRPCClientProxy( String serverName ){
		return  this.serverProxyMap.get( serverName );
	}


	public static class ClientCfg{

		private String serverName;
		private String ip;
		private int port;
		private int num;
		private int maxFrameLength = 20480;


		public String getServerName() {
			return serverName;
		}

		public void setServerName(String serverName) {
			this.serverName = serverName;
		}

		public String getIp() {
			return ip;
		}

		public void setIp(String ip) {
			this.ip = ip;
		}

		public int getPort() {
			return port;
		}

		public void setPort(int port) {
			this.port = port;
		}

		public int getNum() {
			return num;
		}

		public void setNum(int num) {
			this.num = num;
		}

		public int getMaxFrameLength() {
			return maxFrameLength;
		}

		public void setMaxFrameLength(int maxFrameLength) {
			this.maxFrameLength = maxFrameLength;
		}
	}

}
